package com.myself.storm_test;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

@SuppressWarnings("serial")
public class RandomSentenceSpout extends BaseRichSpout {
	private Integer count = 0;
	SpoutOutputCollector _collector;
	Random _rand;

	public void open(Map conf, TopologyContext context,
			SpoutOutputCollector collector) {
		_collector = collector;
		_rand = new Random();
	}

	public void nextTuple() {
		Utils.sleep(100);
		String[] sentences = new String[] { "the cow jumped over the moon",
				"an apple a day keeps the doctor away",
				"four score and seven years ago",
				"snow white and the seven dwarfs", "i am at two with nature",
				"i am at two with nature" };
		// String[]{"{\"a1\":\"12\",\"b1\":\"112\",\"c1\":\"132\",\"d1\":\"134\"}"};
		if (count == sentences.length) {
			System.out.println(count + " try to emit tuple by success_stream");
			_collector.emit("SUCCESS_STREAM", new Values(sentences[0]));
			count++;
		} else if (count < sentences.length) {
			_collector.emit(new Values(sentences[count]));
			count++;
		}

	}

	@Override
	public void ack(Object id) {
	}

	@Override
	public void fail(Object id) {
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word"));
		declarer.declareStream("SUCCESS_STREAM", new Fields("word"));
	}

}